package com.chb.flink.func

import com.chb.flink.source.StationLog
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.util.Collector

/**
 * 把呼叫成功的Stream（主流）和不成功的Stream（侧流）分别输出。
 */
object TestSideOutputStream {


    import org.apache.flink.streaming.api.scala._

    //侧输出流首先需要定义一个流的标签 , 此处需要将隐式转换放在前面
    var notSuccessTag = new OutputTag[StationLog]("not_success")

    def main(args: Array[String]): Unit = {
        //初始化Flink的Streaming（流计算）上下文执行环境

        val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
        streamEnv.setParallelism(1)

        //读取文件数据
        val data = streamEnv.readTextFile(getClass.getResource("/station.log").getPath)
            .map(line => {
                var arr = line.split(",")
                new StationLog(arr(0).trim, arr(1).trim, arr(2).trim, arr(3).trim, arr(4).trim.toLong, arr(5).trim.toLong)
            })

        val result = data.process(new CreateSideOutputStream(notSuccessTag))
        result.print("主流")

        // 输出测流
        // 测流一定需要通过主流获取
        val sideOutputStream = result.getSideOutput(notSuccessTag)
        sideOutputStream.print("测流")



        streamEnv.execute()

    }

    class CreateSideOutputStream(outputTag: OutputTag[StationLog]) extends ProcessFunction[StationLog, StationLog] {
        override def processElement(i: StationLog, context: ProcessFunction[StationLog, StationLog]#Context,
                                    collector: Collector[StationLog]): Unit = {
            if (i.callType.equals("success")) {
                collector.collect(i)  // 输出主流
            } else {
                context.output(outputTag, i)
            }

        }
    }

}